#include "MprpcChannel.h"
#include "MprpcApplication.h"
#include "User.pb.h"
#include "MprpcHeader.pb.h"

#include <sys/types.h>
#include <sys/socket.h>
#include <cerrno>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>

#define RECVBUFFSIZE 1024

// 所有通过stub代理对象调用的rpc方法，都构成多态，走到了这里，统一实现rpc方法调用请求的序列化和网络发送
void MprpcChannel::CallMethod(const google::protobuf::MethodDescriptor *method,
                              google::protobuf::RpcController *controller,
                              const ::google::protobuf::Message *request,
                              google::protobuf::Message *response,
                              google::protobuf::Closure *done)
{
    const google::protobuf::ServiceDescriptor *serviceDesc = method->service();

    // 框架对 rpc请求的序列化
    // header_size + service_name method_name args_size + args
    std::string service_name = serviceDesc->name(); // 调用的服务对象名
    std::string method_name = method->name();       // 调用的函数方法名
    std::string rpc_header_str;
    uint32_t header_size;

    uint32_t args_size = 0;
    std::string args_str;

    if (request->SerializePartialToString(&args_str)) // 序列化出调用的方法参数
    {
        args_size = args_str.size();
    }
    else
    {
        controller->SetFailed("request serialize error!");
        return;
    }

    mprpc::MprpcHeader rpcHeader;
    rpcHeader.set_service_name(service_name);
    rpcHeader.set_method_name(method_name);
    rpcHeader.set_args_size(args_size);

    if (rpcHeader.SerializePartialToString(&rpc_header_str))
    {
        header_size = rpc_header_str.size();
    }
    else
    {
        controller->SetFailed("rpcHeader serialize error!");
        return;
    }

    // 组合成符合规则结构的要发送的字符串
    std::string send_str;
    send_str.insert(0, std::string((char *)&header_size, 4)); // TO PONDER
    send_str += rpc_header_str;
    send_str += args_str;

    // 打印调试信息
    // std::cout << "======================================================" << std::endl;
    // std::cout << "send_str: " << send_str << std::endl;
    // std::cout << "header_size: " << header_size << std::endl;
    // std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    // std::cout << "service_name: " << service_name << std::endl;
    // std::cout << "method_name: " << method_name << std::endl;
    // std::cout << "args_size: " << args_size << std::endl;
    // std::cout << "args_str: " << args_str << std::endl;
    // std::cout << "======================================================" << std::endl;

    int clientfd = socket(AF_INET, SOCK_STREAM, 0);
    if (clientfd == -1)
    {
        char err_buf[512] = {0};
        snprintf(err_buf, 512, "clientfd create error, errno:%d", errno);
        controller->SetFailed(err_buf);
        return;
    }

    // // 获取单例对象，得到配置文件中的地址
    // std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcServerIp");
    // uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcServerPort").c_str());

    // 使用zookeeper
    ZkClient zkclient;
    zkclient.Satrt();   // 建立会话

    // 组装节点路径
    std::string znode_path = "/" + service_name + "/" + method_name;
    // 读取节点数据 ip:port
    std::string znode_data = zkclient.GetData(znode_path.c_str());
    int div = znode_data.find(":");
    std::string ip = znode_data.substr(0, div);
    uint16_t port = atoi(znode_data.substr(div + 1).c_str());

    struct sockaddr_in server_addr;
    bzero(&server_addr, sizeof server_addr);
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(port);
    server_addr.sin_addr.s_addr = inet_addr(ip.c_str());

    // 连接rpc服务节点
    int n = connect(clientfd, (struct sockaddr *)&server_addr, sizeof server_addr);
    if (n == -1)
    {
        close(clientfd);
        char err_buf[512] = {0};
        snprintf(err_buf, 512, "connect error, errno::%d", errno);
        controller->SetFailed(err_buf);
        return;
    }

    // 发送给rpc服务节点
    n = send(clientfd, send_str.c_str(), send_str.size(), 0);
    if (n == -1)
    {
        close(clientfd);
        char err_buf[512] = {0};
        snprintf(err_buf, 512, "send error, errno::%d", errno);
        controller->SetFailed(err_buf);
        return;
    }

    // 接收rpc请求的响应
    char recv_buf[RECVBUFFSIZE] = {0};
    n = recv(clientfd, recv_buf, RECVBUFFSIZE - 1, 0);
    if (n == -1)
    {
        close(clientfd);
        char err_buf[512] = {0};
        snprintf(err_buf, 512, "recv error, errno::%d", errno);
        controller->SetFailed(err_buf);
        return;
    }

    // rpc响应的反序列化
    // recv_buf不一定满
    // std::string response_str(recv_buf, 0, n);
    // if (!response->ParseFromString(response_str))
    if (!response->ParseFromArray(recv_buf, n))
    {
        close(clientfd);
        char err_buf[512] = {0};
        snprintf(err_buf, 512, "response parse error, response_str:%s ", recv_buf);
        controller->SetFailed(err_buf);
        return;
    }
    close(clientfd);
}